package flink_p1

import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.streaming.api.functions.source.{ParallelSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

object FlinkTest_06_CustomSource {

  def main(args: Array[String]): Unit = {


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    //SourceFunction 只支持单线程的产生数据
    val customSource: SourceFunction[String] = new SourceFunction[String] {
      var isStop = true

      override def run(ctx: SourceFunction.SourceContext[String]): Unit = {
        while (isStop) {
          ctx.collect("this is a message")
        }
      }

      override def cancel(): Unit = {

        isStop = false

      }
    }


    /**
     * 报错：
     * Exception in thread "main" java.lang.IllegalArgumentException: Source: 1 is not a parallel source
     */
    //    env.addSource(customSource).setParallelism(2).print()

    // it`s ok
    //    env.addSource(customSource).print()


    val ds2: ParallelSourceFunction[String] = new ParallelSourceFunction[String] {

      var isStop = true

      override def run(ctx: SourceFunction.SourceContext[String]): Unit = {
        while (isStop) {
          ctx.collect("this is a message")
        }
      }

      override def cancel(): Unit = {

        isStop = false

      }
    }


    // it`s ok
    env.addSource(ds2).setParallelism(2).print()


    env.execute("custom source")


  }

}
